/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *    Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.reader;

import com.dtdream.emr.transmission.job.Job;
import com.dtdream.emr.transmission.transfer.TransferNode;
import com.dtdream.emr.transmission.data.Record;
import com.dtdream.emr.transmission.exception.TransmissionException;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_READER_INTERRUPTED;
import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_READER_READ;

public class BufferedReader implements Reader {

    private Job job;
    private final int id;

    private LinkedBlockingQueue queue = new LinkedBlockingQueue(2000);
    private TransferNode transferNode;

    private volatile boolean finished = false;
    private volatile TransmissionException emrException;

    private volatile boolean closed = false;


    public BufferedReader(Job job, int id, TransferNode node) {
        this.job = job;
        this.id = id;
        this.transferNode = node;
    }

    public int getId() {
        return id;
    }

    public void buffer(List<Record> records) throws InterruptedException {
        synchronized (this) {
            for (Record record : records) {
                queue.put(record);
            }
            this.notify();
        }
    }

    private static final long TIMEOUT_MILLS = 1000 * 30;

    private Record pollRecord() throws TransmissionException {
        if (!queue.isEmpty()) {
            return (Record) queue.poll();
        }

        if (null != emrException) {
            throw emrException;
        }

        return null;
    }

    public Record read() throws TransmissionException {
        Record record = pollRecord();
        if (null == record) {
            synchronized (this) {
                if (null == emrException && queue.isEmpty() && !finished) {
                    transferNode.readAsync(job.getJobId(), id);

                    try {
                        this.wait(TIMEOUT_MILLS);
                    } catch (InterruptedException e) {
                        throw new TransmissionException(EXCEPTION_SDK_READER_INTERRUPTED);
                    }
                }
            }
            record = pollRecord();
        }

        return record;
    }

    public void finish() {
        synchronized (this) {
            finished = true;
            this.notify();
        }
    }

    public synchronized void close() throws TransmissionException {
        closed = true;
        job.transferClient.closeReader(job.getJobId(), id, this.transferNode);
    }

    public boolean isClosed() {
        return closed;
    }

    @Override
    public void setError(TransmissionException cause) {
        synchronized (this) {
            this.emrException = new TransmissionException(EXCEPTION_SDK_READER_READ, cause);
            this.notify();
        }
    }
}
